// Copyright tang.  All rights reserved.
// https://gitee.com/inrgihc/dbswitch
//
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package middle.ground.service;

import cn.hutool.core.exceptions.ExceptionUtil;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.extern.slf4j.Slf4j;
import middle.ground.dao.AssignmentConfigDAO;
import middle.ground.dao.AssignmentJobDAO;
import middle.ground.dao.AssignmentTaskDAO;
import middle.ground.dbswitch.data.config.DbswichProperties;
import middle.ground.dbswitch.data.service.MigrationService;
import middle.ground.entity.AssignmentConfigEntity;
import middle.ground.entity.AssignmentJobEntity;
import middle.ground.entity.AssignmentTaskEntity;
import middle.ground.type.JobStatusEnum;
import middle.ground.util.JsonUtils;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.InterruptableJob;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobKey;
import org.quartz.PersistJobDataAfterExecution;
import org.quartz.UnableToInterruptJobException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.sql.Timestamp;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>
 * 如果你使用了@PersistJobDataAfterExecution注解，则强烈建议你同时使用@DisallowConcurrentExecution注
 * 解，因为当同一个job（JobDetail）的两个实例被并发执行时，由于竞争，JobDataMap中存储的数据很可能是不确定的。
 * </p>
 */
@Slf4j
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public class JobExecutorService extends QuartzJobBean implements InterruptableJob {

	public final static String GROUP = "dbswitch";
	public final static String TASK_ID = "taskId";
	public final static String SCHEDULE = "schedule";

	// 相同taskId的任务限制并发执行的粒度锁缓存对象
	private static Cache<String, ReentrantLock> mutexes = CacheBuilder.newBuilder()
			.expireAfterWrite(24 * 60L, TimeUnit.MINUTES)
			.build();

	/**
	 * 作为一个是否被中断的标识
	 */
	private boolean interrupted = false;

	/**
	 * 记录当前线程
	 */
	private Thread currentThread;

	/**
	 * 因为在QuartzConfig中进行了注入配置，所以 Quartz会将数据注入到jobKey变量中
	 */
	private String taskId;

	@Autowired
	private AssignmentTaskDAO assignmentTaskDAO;
	@Autowired
	private AssignmentConfigDAO assignmentConfigDAO;
	@Autowired
	private AssignmentJobDAO assignmentJobDAO;

	/**
	 * 实现setter方法，Quartz会给成员变量taskId注入值
	 *
	 * @param taskId Task的ID
	 */
	public void setTaskId(String taskId) {
		this.taskId = taskId;
	}

	@Override
	public void interrupt() {
		log.info("Quartz Schedule Task job is interrupting : taskId={} ", taskId);
		interrupted = true;
		currentThread.interrupt();
	}

	@Override
	public void executeInternal(JobExecutionContext context) {
		currentThread = Thread.currentThread();
		JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
		if (interrupted) {
			log.info("Quartz task id:{} interrupted", jobDataMap.getLong(TASK_ID));
			return;
		}

		JobKey key = context.getJobDetail().getKey();
		Long taskId = jobDataMap.getLongValue(TASK_ID);
		Integer schedule = jobDataMap.getIntValue(SCHEDULE);
		AssignmentJobEntity assignmentJobEntity = assignmentJobDAO
				.newAssignmentJob(taskId, schedule, key.getName());

		try {
			ReentrantLock lock = mutexes.get(taskId.toString(), ReentrantLock::new);
			while (!lock.tryLock(1, TimeUnit.SECONDS)) {
				TimeUnit.SECONDS.sleep(1);
			}

			try {
				log.info("Execute Quartz Job, and task id is : {} , job id is: {}", taskId,
						assignmentJobEntity.getId());

				AssignmentTaskEntity task = assignmentTaskDAO.getById(taskId);
				AssignmentConfigEntity assignmentConfigEntity = assignmentConfigDAO
						.getByAssignmentTaskId(task.getId());

				log.info("Execute Assignment [taskId={}],Task Name: {} ,configuration properties：{}",
						task.getId(),
						task.getName(),
						task.getContent());

				try {
					DbswichProperties properties = JsonUtils.toBeanObject(
							task.getContent(), DbswichProperties.class);
					//zrx 不判断是否是第一次
					/*if (!assignmentConfigEntity.getFirstFlag()) {
						properties.getTarget().setTargetDrop(false);
						properties.getTarget().setChangeDataSync(true);
					}*/

					MigrationService mainService = new MigrationService(properties);
					if (interrupted) {
						log.info("Quartz task id:{} interrupted", jobDataMap.getLong(TASK_ID));
						return;
					}

					// 实际执行JOB
					mainService.run();

					if (assignmentConfigEntity.getFirstFlag()) {
						AssignmentConfigEntity config = new AssignmentConfigEntity();
						config.setId(assignmentConfigEntity.getId());
						//zrx 注释掉
						/*config.setTargetDropTable(Boolean.FALSE);
						config.setTargetDataSync(Boolean.TRUE);*/
						config.setFirstFlag(Boolean.FALSE);
						assignmentConfigDAO.updateSelective(config);
					}

					assignmentJobEntity.setStatus(JobStatusEnum.PASS.getValue());
					log.info("Execute Assignment Success [taskId={},jobId={}],Task Name: {}",
							task.getId(), assignmentJobEntity.getId(), task.getName());
				} catch (Throwable e) {
					assignmentJobEntity.setStatus(JobStatusEnum.FAIL.getValue());
					assignmentJobEntity.setErrorLog(ExceptionUtil.stacktraceToString(e));
					log.info("Execute Assignment Failed [taskId={},jobId={}],Task Name: {}",
							task.getId(), assignmentJobEntity.getId(), task.getName(), e);
				} finally {
					assignmentJobEntity.setFinishTime(new Timestamp(System.currentTimeMillis()));
					assignmentJobDAO.updateSelective(assignmentJobEntity);
				}
			} finally {
				lock.unlock();
			}
		} catch (ExecutionException | InterruptedException e) {
			log.error("定时任务执行出错：" + e.getMessage(), e);
		}

	}

}
